#include "ul_memory_pool.h"
#include <string.h>
#include <new>
#include <errno.h>
#include <sched.h>
#include "ul_define.h"
using namespace LinuxNetworkUtility;
namespace
{
  const int64_t BUCKET_BLOCK_SIZE_ARRAY[]=
  {
    1024,
    (1<<1)*1024ll,
    (1<<2)*1024ll,
    (1<<3)*1024ll,
    (1<<4)*1024ll,
    (1<<5)*1024ll,
    (1<<6)*1024ll,
    (1<<7)*1024ll,
    (1<<8)*1024ll,
    (1<<9)*1024ll,
    (1<<10)*1024ll,
    (1<<11)*1024ll,
    (1<<12)*1024ll,
    (1<<13)*1024ll,
    (1<<14)*1024ll,
    (1<<15)*1024ll,
    (1<<16)*1024ll,
    (1<<17)*1024ll,
    (1<<18)*1024ll,
    (1<<19)*1024ll,
    (1<<20)*1024ll,
    (1<<21)*1024ll,
    (1<<22)*1024ll,
    (1<<23)*1024ll,
    (1<<24)*1024ll
  };

  /// optimize out memory fragment
  static const int32_t MINIMUM_BLOCK_SIZE_POWER = 5;

  int32_t get_bucket_index(int64_t size)
  {
    int32_t result = -1;
    if (size > 0)
    {
      for (int bucket_idx = MINIMUM_BLOCK_SIZE_POWER; 
          bucket_idx < UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX; 
          bucket_idx ++)
      {
        if (BUCKET_BLOCK_SIZE_ARRAY[bucket_idx] -static_cast<int64_t>(sizeof(UlMemoryBlock)) > size)
        {
          result = bucket_idx;
          break;
        }
      }
    }
    return result;
  }
}

LinuxNetworkUtility::UlMemoryPool::UlMemoryPool(UlTimer & time)
:time_(time)
{
  memset(memory_block_num_,0x00,sizeof(memory_block_num_));
  memory_size_handled_ = 0;
  memory_size_in_use_ = 0;
  pthread_mutex_init(&mutex_,NULL);
}


LinuxNetworkUtility::UlMemoryPool::~UlMemoryPool()
{
  DLink *blk_it = NULL;
  UlMemoryBlock *blk = NULL;
  pthread_mutex_lock(&mutex_);
  for (int32_t bucket_idx = 0; bucket_idx <= UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX; bucket_idx ++)
  {
    blk_it = memory_block_bucket_[bucket_idx].next();
    while (blk_it != memory_block_bucket_ + bucket_idx)
    {
      blk = CONTAINING_RECORD(blk_it, UlMemoryBlock,dlink_);
      if (blk->check_validity(bucket_idx))
      {
        blk_it = blk_it->next();
        if (blk->block_in_use_)
        {
          UL_LOG_ERROR("block in use when memory pool was destroyed [addr:%p]",blk);
        }
        else
        {
          blk->free(time_, bucket_idx);
          delete [] reinterpret_cast<char*>(blk);
          memory_size_handled_ -= BUCKET_BLOCK_SIZE_ARRAY[bucket_idx];
        } 
      }
      else
      {
        UL_LOG_ERROR("memory block validity check fail [addr:%p]", blk);
        break;
      }
    }
  }
  pthread_mutex_unlock(&mutex_);
  pthread_mutex_destroy(&mutex_);
}


UlMemoryBlock * LinuxNetworkUtility::UlMemoryPool::malloc(const int64_t size, 
                                                          const UlMemoryFreeFlag free_flag,  int32_t mod_id)
{
  int err = 0;
  UlMemoryBlock * result = NULL;
  int32_t bucket_idx = -1;
  bucket_idx = get_bucket_index(size);
  if (bucket_idx < 0 || bucket_idx > UL_MAX_BLOCK_SIZE_POWER_LEVEL)
  {
    UL_LOG_WARNING("param error [size:%lld,max_block_size:%lld]", size, UL_MAX_BLOCK_SIZE);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  if (0 == err)
  {
    pthread_mutex_lock(&mutex_);
    if (!memory_block_bucket_[bucket_idx].is_empty())
    {
      result = CONTAINING_RECORD(memory_block_bucket_[bucket_idx].next(),UlMemoryBlock, dlink_) ;
      result->dlink_.remove();
      memory_size_in_use_ += BUCKET_BLOCK_SIZE_ARRAY[bucket_idx];
      result->malloc(free_flag,time_,mod_id);
    }
    else
    {
      result = reinterpret_cast<UlMemoryBlock*>(new(std::nothrow)char[BUCKET_BLOCK_SIZE_ARRAY[bucket_idx]]);
      if (NULL == result)
      {
        UL_LOG_WARNING("fail to allocate memory from system [errno:%d]",errno);
        err = ERRNO2ERR(errno);
      }
      else
      {
        result->init(bucket_idx, free_flag, time_, mod_id);
        memory_size_in_use_ += BUCKET_BLOCK_SIZE_ARRAY[bucket_idx];
        memory_size_handled_ += BUCKET_BLOCK_SIZE_ARRAY[bucket_idx];
      }
    }
    pthread_mutex_unlock(&mutex_);
  }
  return result;
}


void LinuxNetworkUtility::UlMemoryPool::free(UlMemoryBlock *block)
{
  DLink blk_head;
  DLink *blk_it = NULL;
  UlMemoryBlock *lblock = NULL;
  if (NULL != block)
  {
    if (block->check_validity(block->block_size_power_))
    {
      block->dlink_.insert_prev(blk_head);
      blk_it = blk_head.next();
      pthread_mutex_lock(&mutex_);
      while (blk_it != &blk_head)
      {
        lblock = CONTAINING_RECORD(blk_it, UlMemoryBlock, dlink_);
        if (lblock->check_validity(lblock->block_size_power_))
        {
          blk_it = blk_it->next();
          lblock->free(time_, lblock->block_size_power_);
          memory_size_in_use_ -= BUCKET_BLOCK_SIZE_ARRAY[lblock->block_size_power_];
          memory_block_bucket_[lblock->block_size_power_].insert_next(lblock->dlink_);
        }
        else
        {
          UL_LOG_ERROR("memory block validity check fail [addr:%p]", lblock);
          break;
        }
      }
      pthread_mutex_unlock(&mutex_);
    }
    else
    {
      UL_LOG_ERROR("memory block validity check fail [addr:%p]", block);
    }
  }
}


void LinuxNetworkUtility::UlMemoryPool::shrink()
{
  DLink *blk_it = NULL;
  UlMemoryBlock *blk = NULL;
  int64_t now = time_.get_cur_time_ul_minute();
  int64_t gc_memory_size = 0;
  pthread_mutex_lock(&mutex_);
  for (int32_t bucket_idx = 0; bucket_idx < UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX; bucket_idx ++)
  {
    blk_it = memory_block_bucket_[bucket_idx].prev();
    while (blk_it != memory_block_bucket_ + bucket_idx)
    {
      blk = CONTAINING_RECORD(blk_it, UlMemoryBlock,dlink_);
      if (blk->check_validity(bucket_idx))
      {
        blk_it = blk_it->prev();
        if (blk->block_in_use_)
        {
          UL_LOG_ERROR("block in use after it was reverted to memory pool [addr:%p]",blk);
        }
        else if (now - blk->last_access_time_ >= UL_MEMORY_POOL_GC_BLOCK_AGE_IN_UL_MINUTE)
        {
          blk->free(time_, bucket_idx);
          delete [] reinterpret_cast<char*>(blk);
          memory_size_handled_ -= BUCKET_BLOCK_SIZE_ARRAY[bucket_idx];
          gc_memory_size += BUCKET_BLOCK_SIZE_ARRAY[bucket_idx];
        }
        else
        {
          break;
        }
      }
      else
      {
        UL_LOG_ERROR("memory block validity check fail [addr:%p]", blk);
        break;
      }
    }
  }
  pthread_mutex_unlock(&mutex_);
  UL_LOG_INFO("memory pool gced memory size :%lld", gc_memory_size);
}

LinuxNetworkUtility::UlThreadMemoryPool::UlThreadMemoryPool(UlMemoryPool &global_pool, 
                                                            UlMemPoolModSet &mod_set, 
                                                            UlTimer &time,
                                                            int64_t gc_frequency_us)
:mod_set_(mod_set),time_(time),global_pool_(global_pool)
{
  memset(memory_block_num_,0x00,sizeof(memory_block_num_));
  memory_size_handled_ = 0;
  memory_size_in_use_ = 0;
  cur_used_block_it_ = memory_block_bucket_ + UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX;
  gc_frequency_us_ = gc_frequency_us;
  if (gc_frequency_us <= 0)
  {
    gc_frequency_us_ = 10000;
  }
  last_gc_time_ = time_.get_cur_time_ul_minute();
  local_cache_hit_times_ = 0;
  local_cache_missed_times_ = 0;
}

LinuxNetworkUtility::UlThreadMemoryPool::~UlThreadMemoryPool()
{
  DLink *blk_it = NULL;
  UlMemoryBlock *blk = NULL;
  for (int32_t bucket_idx = 0; bucket_idx < UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX; bucket_idx ++)
  {
    blk_it = memory_block_bucket_[bucket_idx].next();
    while (blk_it != memory_block_bucket_ + bucket_idx)
    {
      blk = CONTAINING_RECORD(blk_it, UlMemoryBlock,dlink_);
      if (blk->check_validity(bucket_idx))
      {
        blk_it = blk_it->next();
        if (blk->block_in_use_)
        {
          UL_LOG_ERROR("block in use when memory pool was destroyed [addr:%p]",blk);
        }
        else
        {
          blk->dlink_.remove();
          global_pool_.free(blk);
          memory_size_handled_ -= BUCKET_BLOCK_SIZE_ARRAY[bucket_idx];
        } 
      }
      else
      {
        UL_LOG_ERROR("memory block validity check fail [addr:%p]", blk);
        break;
      }
    }
  }

  for (;;)
  {
    blk_it = memory_block_bucket_[UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX].next();
    while (blk_it != memory_block_bucket_ + UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX)
    {
      blk = CONTAINING_RECORD(blk_it, UlMemoryBlock,dlink_);
      if (blk->check_validity(blk->block_size_power_))
      {
        blk_it = blk_it->next();
        if (blk->block_in_use_)
        {
        }
        else
        {
          blk->dlink_.remove();
          global_pool_.free(blk);
          memory_size_handled_ -= BUCKET_BLOCK_SIZE_ARRAY[blk->block_size_power_];
        } 
      }
      else
      {
        UL_LOG_ERROR("memory block validity check fail [addr:%p]", blk);
        break;
      }
    }
    if (memory_block_bucket_[UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX].is_empty())
    {
      break;
    }
    else
    {
      sched_yield();
    }
  }

  UL_LOG_MONITOR("local_cache_hit_times:%lld,local_cache_missed_times:%lld", local_cache_hit_times_,
                 local_cache_missed_times_);
}


void LinuxNetworkUtility::UlThreadMemoryPool::gc_()
{
  static const int32_t REVERT_TO_ALLOCATED_THREAD_LIST_TRAVERSE_ITEM_NUM = 2048;
  int64_t now = time_.get_cur_time_ul_minute();
  DLink * blk_it = NULL;
  UlMemoryBlock *blk = NULL;
  /// first gc all block which is not active in the last ul minute
  for (int32_t bucket_idx = 0; bucket_idx < UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX; bucket_idx ++)
  {
    blk_it = memory_block_bucket_[bucket_idx].prev();
    while (blk_it != memory_block_bucket_ + bucket_idx)
    {
      blk = CONTAINING_RECORD(blk_it, UlMemoryBlock, dlink_);
      blk_it = blk_it->next();
      if (blk->check_validity(blk->block_size_power_))
      {
        if (blk->block_in_use_)
        {
          UL_LOG_ERROR("block was in use after revert to memory pool [addr:%p]", blk);
          break;
        }
        else if (now - blk->last_access_time_ >= UL_MEMORY_POOL_GC_BLOCK_AGE_IN_UL_MINUTE)
        {
          blk->dlink_.remove();
          memory_size_handled_ -= BUCKET_BLOCK_SIZE_ARRAY[blk->block_size_power_];
          global_pool_.free(blk);
        }
        else
        {
          break;
        }
      }
      else
      {
        UL_LOG_ERROR("memory block validity check fail [addr:%p]", blk);
      }
    }
  }
  /// second, gc all memory return to allocated thread's memory pool
  cur_used_block_it_ = cur_used_block_it_->next();
  for (int32_t traversed_num = 0; 
      (traversed_num < REVERT_TO_ALLOCATED_THREAD_LIST_TRAVERSE_ITEM_NUM) 
      && (cur_used_block_it_ != memory_block_bucket_+UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX);
      traversed_num ++)
  {
    blk = CONTAINING_RECORD(cur_used_block_it_,UlMemoryBlock,dlink_);
    if (blk->check_validity(blk->block_size_power_))
    {
      cur_used_block_it_ = cur_used_block_it_->next();
      if (!blk->block_in_use_)
      {
        blk->free(time_, blk->block_size_power_);
        memory_size_in_use_ -= BUCKET_BLOCK_SIZE_ARRAY[blk->block_size_power_];
        memory_block_bucket_[blk->block_size_power_].insert_next(blk->dlink_);
      }
    }
    else
    {
      UL_LOG_ERROR("memory block validity check fail [addr:%p]", blk);
      break;
    }
  }
  last_gc_time_ = now;
}

void LinuxNetworkUtility::UlThreadMemoryPool::free(void *ptr)
{
  UlMemoryBlock *blk = NULL;
  if (NULL != ptr)
  {
    blk = reinterpret_cast<UlMemoryBlock*>(reinterpret_cast<char*>(ptr) - sizeof(UlMemoryBlock));
    if (blk->check_validity(blk->block_size_power_))
    {
      switch (blk->free_indicator_flags_)
      {
      case FREE_TO_THREAD_POOL:
        if (blk->last_access_thead_id_ != gettid())
        {
          UL_LOG_ERROR("usage error [free_flag:%d,last_access_tid:%u,cur_tid:%u]", 
                       blk->free_indicator_flags_, blk->last_access_thead_id_, gettid());
        }
        else
        {
          blk->free(time_, blk->block_size_power_);
          memory_size_in_use_ -= BUCKET_BLOCK_SIZE_ARRAY[blk->block_size_power_];
          memory_block_bucket_[blk->block_size_power_].insert_next(blk->dlink_);
        }
        break;
      case FREE_TO_GLOBAL_POOL:
        memory_size_in_use_ -= BUCKET_BLOCK_SIZE_ARRAY[blk->block_size_power_];
        memory_size_handled_ -= BUCKET_BLOCK_SIZE_ARRAY[blk->block_size_power_];
        blk->last_access_thead_id_ = gettid();
        global_pool_.free(blk);
        break;
      case FREE_TO_ALLOCATED_THREAD_POOL:
        blk->block_in_use_ = 0;
        blk->last_access_thead_id_ = gettid();
        break;
      default:
        UL_LOG_ERROR("memory block validity check fail [addr:%p,flag:%d]", blk, blk->free_indicator_flags_);
      }
      mod_set_.free(BUCKET_BLOCK_SIZE_ARRAY[blk->block_size_power_],blk->mod_id_);
    }
    else
    {
      UL_LOG_ERROR("memory block validity check fail [addr:%p]", blk);
    }
  }
  /*
  if (time_.get_cur_time_ul_minute() - last_gc_time_ > gc_frequency_ul_minute_)
  {
    gc_();
  } 
  */
}

void *LinuxNetworkUtility::UlThreadMemoryPool::malloc(const int64_t size, const int32_t mod_id, 
                                                      int64_t * got_size, const UlMemoryFreeFlag flag)
{
  int err = 0;
  void *result = NULL;
  UlMemoryBlock * result_block = NULL;
  int32_t bucket_idx = -1;
  bucket_idx = get_bucket_index(size);
  if (bucket_idx < 0 || bucket_idx > UL_MAX_BLOCK_SIZE_POWER_LEVEL)
  {
    UL_LOG_WARNING("param error [size:%lld,max_block_size:%lld]", size, UL_MAX_BLOCK_SIZE);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }
  else if (flag < FREE_TO_THREAD_POOL || flag > FREE_TO_ALLOCATED_THREAD_POOL)
  {
    UL_LOG_WARNING("param error, unrecogonized flag [flag:%ld]", flag);
    errno = EINVAL;
    err = ERRNO2ERR(errno);
  }

  if (0 == err && time_.get_cur_time_us() - last_gc_time_ > gc_frequency_us_)
  {
    gc_();
  }

  if (0 == err)
  {
    if (!memory_block_bucket_[bucket_idx].is_empty())
    {
      result_block = CONTAINING_RECORD(memory_block_bucket_[bucket_idx].next(),UlMemoryBlock, dlink_) ;
      result_block->dlink_.remove();
      memory_size_in_use_ += BUCKET_BLOCK_SIZE_ARRAY[bucket_idx];
      result_block->malloc(flag,time_,mod_id);
      local_cache_hit_times_ ++;
    }
    else
    {
      result_block = global_pool_.malloc(size,flag,mod_id);
      if (NULL == result_block)
      {
        UL_LOG_WARNING("fail to allocate memory from global memory pool [errno:%d]",errno);
        err = ERRNO2ERR(errno);
      }
      else
      {
        memory_size_in_use_ += BUCKET_BLOCK_SIZE_ARRAY[bucket_idx];
        memory_size_handled_ += BUCKET_BLOCK_SIZE_ARRAY[bucket_idx];
        local_cache_missed_times_ ++;
      }
    }
  }
  if (NULL != got_size)
  {
    *got_size = 0;
  }
  if (NULL != result_block)
  {
    switch (flag)
    {
    case FREE_TO_THREAD_POOL:
      break;
    case FREE_TO_ALLOCATED_THREAD_POOL:
      memory_block_bucket_[UL_REVERT_TO_THREAD_POOL_BLOCK_LIST_IDX].insert_next(result_block->dlink_);
      break;
    case FREE_TO_GLOBAL_POOL:
      break;
    default:
      /// will not reach here
      break;
    }
    result = result_block->buf_;
    mod_set_.malloc(BUCKET_BLOCK_SIZE_ARRAY[result_block->block_size_power_],mod_id);
    if (NULL != got_size)
    {
      *got_size = BUCKET_BLOCK_SIZE_ARRAY[result_block->block_size_power_]-sizeof(UlMemoryBlock);
    }
  }
  return result;
}
